Book Case Study - IoT Sensor Streaming

This guide covers the classic IoT Sensor Streaming case study from Chapter 8 of the official O'Reilly book Learning Spark (2nd Edition) by Jules S. Damji, Brooke Wenig, Tathagata Das, and Denny Lee. It demonstrates how to leverage Spark's Structured Streaming API to ingest continuous streams of JSON files, enforce strict event-time schemas, run windowed aggregations, and apply Watermarks to manage late-arriving IoT data.

The Scenario

Suppose we have millions of IoT sensor devices deployed worldwide. Each device continuously generates metrics containing a device ID, temperature readings, battery levels, network signal strength, location codes, and a creation timestamp. These events are dumped as JSON text files into a landing directory.

By applying Structured Streaming, we will:

1. Define a programmatic schema to safely read incoming JSON files in a streaming fashion.
2. Build a sliding event-time window aggregation to compute the average temperature and lowest battery level across 10-minute intervals, sliding every 5 minutes.
3. Apply a 10-minute event-time Watermark to allow Spark's state store to discard historical window metadata, preventing out-of-memory crashes on continuous streams.
4. Write the results to active stream sinks (Console/Memory) with triggers.

Programmatic Streaming Schema

A strict, programmatic schema is mandatory for file-based streaming sources because Spark does not infer schemas on empty directories or dynamically landing files.

from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType
)

# Explicit schema representing incoming IoT device sensor payloads
iot_schema = StructType([
    StructField("device_id", IntegerType(), True),
    StructField("device_name", StringType(), True),
    StructField("ip", StringType(), True),
    StructField("cca2", StringType(), True),
    StructField("cca3", StringType(), True),
    StructField("cn", StringType(), True),
    StructField("temp", DoubleType(), True),
    StructField("signal", DoubleType(), True),
    StructField("battery_level", IntegerType(), True),
    StructField("c02_level", IntegerType(), True),
    StructField("timestamp", TimestampType(), True)
])

PySpark Script: Structured Streaming IoT Aggregator

Below is the complete, runnable Structured Streaming code. It configures a file stream reader, aggregates windows with watermarking, and runs the streaming query.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window, avg, min

# 1. Initialize SparkSession
spark = SparkSession.builder \
    .appName("IoTSensorStreamingAggregator") \
    .master("local[*]") \
    .getOrCreate()

# Set shuffle partitions to a small number for local stream testing (avoids 200 partition overhead)
spark.conf.set("spark.sql.shuffle.partitions", "4")

# 2. Define the Directory Stream Reader
# In production, this points to an S3/HDFS/ADLS folder or a Kafka stream
input_directory = "iot_devices_landing_zone/"

streaming_df = spark.readStream \
    .format("json") \
    .schema(iot_schema) \
    .option("maxFilesPerTrigger", "1") \
    .load(input_directory)

# 3. Apply Sliding Window and Event-Time Watermarking
# We aggregate values inside a 10-minute window, sliding every 5 minutes.
# We apply a 10-minute Watermark: late data arriving > 10 minutes past the window edge is discarded.
aggregated_stream_df = streaming_df \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(col("timestamp"), "10 minutes", "5 minutes"),
        col("device_name")
    ) \
    .agg(
        avg("temp").alias("avg_temperature"),
        min("battery_level").alias("min_battery")
    )

# 4. Start the Stream and Write to the Console Sink
# We use standard 'Update' mode to output only changed window aggregate metrics
query = aggregated_stream_df.writeStream \
    .format("console") \
    .outputMode("update") \
    .option("truncate", "false") \
    .trigger(processingTime="5 seconds") \
    .start()

# Await termination (keeps the driver process running to monitor the stream)
query.awaitTermination()

Streaming Engine Operations & Watermarking

Structured Streaming processes data by treating the continuous input stream as an unbounded, ever-growing table. The engine slices the stream, computes incremental updates, and maintains a state store:

graph TD
    A["Raw JSON landing zone (Continuous input)"] -->|"readStream"| B["Spark Streaming Query Engine"]
    B -->|"1. Ingest mini-batch"| C["Apply event-time watermark: timestamp - 10 mins"]
    C -->|"2. Discard events older than watermark"| D["Group into 10-min windows sliding by 5 mins"]
    D -->|"3. Look up state store"| E["Update Active Window aggregations in memory"]
    E -->|"4. outputMode: Update"| F["Console / Memory Sink (Batch output)"]

    style A fill:#f5f5f5,stroke:#9e9e9e,stroke-width:2px;
    style C fill:#eff6ff,stroke:#2563eb,stroke-width:2px;
    style E fill:#fef2f2,stroke:#dc2626,stroke-width:2px;
    style F fill:#f0fdf4,stroke:#16a34a,stroke-width:2px;

Watermark Mechanics:

• The Problem: In real-time streams, network delays can cause data to arrive late (out of order). Spark must retain old window groups in memory to add late data. However, if left unchecked, the in-memory window metadata will grow until the application crashes (Out Of Memory).
• The Solution: A Watermark defines how long Spark waits for late-arriving data. In our script (.withWatermark("timestamp", "10 minutes")), Spark tracks the maximum event time seen so far. The watermark threshold is calculated as:

Watermark = Max Event Time Seen - Watermark Delay (10 minutes)

• State Cleanup: Any window whose end timestamp falls below the calculated watermark threshold is permanently finalized, flushed to the sink, and its keys are purged from the memory state store. Any new records that arrive with timestamps older than the watermark are dropped immediately.